<?php

class  TcpServer
{
    //系统支持的最大进程数目
    const MAX_PROCESS = 3;
    //子进程数组
    private $pids = [];
    //网络套接字
    private $socket;
    //主进程id
    private $mpid;

    /**
     * 服务器主进程业务逻辑
     */
    public function run()
    {
        $process = new \Swoole\Process(function () {
            $this->mpid = posix_getpid();
            echo time() . "Master process, pid {$this->mpid}\n";
            $this->socket = stream_socket_server("tcp://127.0.0.1:9503", $errno, $errstr);
            if (!$this->socket) {
                exit("Server start error: $errstr");
            }
            for ($i = 0; $i < self::MAX_PROCESS; $i++) {
                $this->startWorkerProcess();
            }
            echo("Waiting client staart ... \n");
            while (1) {
                foreach ($this->pids as $k => $pid) {
                    if ($pid) {
                        $ret = \Swoole\Process::wait(false);
                        if ($ret) {
                            echo time() . "Worker process $pid exit will start new ... \n";
                            $this->startWorkerProcess();
                            unset($this->pids[$k]);
                        }
                    }
                }
                sleep(1);
            }
        }, false, false);
        \Swoole\Process::daemon();
        $process->start();
    }

    /**
     * 创建worker进程 接受客户端处理
     */
    private function startWorkerProcess()
    {
        $process = new \Swoole\Process(function (\Swoole\Process $worker) {
            $this->acceptClient($worker);
        }, false, false);
        $pid = $process->start();
        $this->pids[] = $pid;
    }

    /**
     * 等待客户端连接并处理
     * @param $worker
     */
    private function acceptClient(&$worker)
    {
        while (1) {
            $conn = stream_socket_accept($this->socket, -1);
            if ($this->onConnect) {
                call_user_func($this->onConnect, $conn);
            }
            $recv = '';
            $buffer = '';
            while (1) {
                $this->checkMpid($worker);
                $buffer = fread($conn, 20);
                if ($buffer == false || $buffer == '') {
                    if ($this->onClose) {
                        call_user_func($this->onClose, $conn);
                    }
                    break;
                }

                $pos = strpos($buffer, "\n");
                if ($pos === false) {
                    $recv .= $buffer;
                } else {
                    $recv .= trim(substr($buffer, 0, $pos + 1));
                    // 如果服务器定义了消息处理回调函数，则在当前连接上将消息传入回调函数并执行该回调
                    if ($this->onMessage) {
                        call_user_func($this->onMessage, $conn, $recv);
                    }
                    if ($recv == 'quit') {
                        echo "Client close connection\n";
                        fclose($conn);
                        break;
                    }

                    $recv = '';

                }

            }
        }
    }

    public function checkMpid(&$worker){
        if (!\Swoole\Process::kill($this->mpid,0)) {
            $worker->exit();
            echo "Master process exited, I [{$worker['pid']}] also quit\n";
        }
    }

}

$server = new TcpServer();

// 定义连接建立回调函数
$server->onConnect = function ($conn) {
    echo "onConnect -- accepted " . stream_socket_get_name($conn, true) . "\n";
};

// 定义收到消息回调函数
$server->onMessage = function ($conn, $msg) {
    echo "onMessage --" . $msg . "\n";
    fwrite($conn, "received " . $msg . "\n");
};

// 定义连接关闭回调函数
$server->onClose = function ($conn) {
    echo "onClose --" . stream_socket_get_name($conn, true) . "\n";
};

// 启动服务器主进程
$server->run();